/**
 * Copyright 2010 Nicholas Alexander Cross
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.dataporter.data;

import org.apache.commons.lang.NullArgumentException;
import org.apache.log4j.Logger;
import org.dataporter.io.DataReader;
import org.dataporter.io.DataWriter;
import org.dataporter.processing.DataProcessingException;
import org.dataporter.processing.DataProcessor;
import org.dataporter.validation.DataValidationException;
import org.dataporter.validation.DataValidator;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Importer<Source, Sink> implements DataImporter<Source, Sink> {

    private static final Logger logger = Logger.getLogger( Importer.class );

    private final DataReader<Source> reader;
    private final DataWriter<Sink> writer;
    private final Map<String, List<DataProcessor>> transformers = new HashMap<String, List<DataProcessor>>();
    private final Map<String, List<DataValidator>> validators = new HashMap<String, List<DataValidator>>();

    public Importer( DataReader<Source> reader, DataWriter<Sink> writer ) {
        this.reader = reader;
        this.writer = writer;

        if ( reader == null )
        {
            throw new NullArgumentException( "reader" );
        }

        if ( writer == null )
        {
            throw new NullArgumentException( "writer" );
        }
    }

    public List<DataLogEntry> importData( Source source, Sink sink ) {
        List<DataLogEntry> importLog = new ArrayList<DataLogEntry>();
        List<DataObject> objects = new ArrayList<DataObject>();

        try
        {
            //todo: read/write progressively to avoid large memory footprint?
            readObjectsForSource( source, objects );

            writeObjectsToSink( sink, importLog, objects );
        }
        catch ( Exception e )
        {
            logger.error( "Import Failure", e );
            importLog.add( new DataLogEntry( "Import Failure with exception: " + e.getClass().getName() + " - " + e.getMessage() ) );
        }

        return importLog;
    }


    protected void readObjectsForSource( Source source, List<DataObject> objects ) throws Exception {
        reader.init( source );
        
        try
        {
            DataObject dataObject = reader.read();

            while ( dataObject != null )
            {
                objects.add( dataObject );
                dataObject = reader.read();
            }
        }
        finally
        {
            reader.close();
        }
    }

    protected void writeObjectsToSink( Sink sink, List<DataLogEntry> importLog, List<DataObject> objects ) throws Exception {
        writer.init( sink );

        try
        {
            for ( DataObject o : objects )
            {

                String entityName = o.getEntityName();
                List<DataProcessor> dataProcessors = transformers.get( entityName );
                List<DataValidator> dataValidators = validators.get( entityName );

                try
                {
                    List<DataUpdate> updates = new ArrayList<DataUpdate>();

                    if ( dataProcessors != null )
                    {
                        for ( DataProcessor processor : dataProcessors )
                        {
                            List<DataUpdate> currentUpdates = processor.process( o );

                            if ( currentUpdates != null )
                            {
                                updates.addAll( currentUpdates );
                            }
                        }
                    }

                    if ( dataValidators != null )
                    {
                        for ( DataValidator validator : dataValidators )
                        {
                            validator.validate( o, updates );
                        }
                    }

                    DataLogEntry entry = null;

                    for ( DataUpdate update : updates )
                    {
                        entry = writer.write( update );
                    }

                    if ( entry != null )
                    {
                        importLog.add( entry );
                    }

                }
                catch ( DataValidationException e )
                {
                    logger.warn( "Validation Failure", e );
                    if (e.getSeverity() != DataValidationException.Severity.Fatal) {
                        importLog.add( e.logEntry() );
                    }
                    else {
                        throw e;
                    }
                }
                catch ( DataProcessingException e )
                {
                    logger.warn( "Transformation Failure", e );
                    importLog.add( e.logEntry() );
                }
            }
        }
        finally
        {
            writer.close();
        }
    }

    public void addProcessor( String entityName, DataProcessor processor ) {
        List<DataProcessor> list = transformers.get( entityName );

        if ( list == null )
        {
            list = new ArrayList<DataProcessor>();
            transformers.put( entityName, list );
        }

        list.add( processor );
    }

    public void addProcessor( String entityName, DataProcessor... dataProcessors ) {
        for ( DataProcessor t : dataProcessors )
        {
            addProcessor( entityName, t );
        }
    }

    public void addValidator( String entityName, DataValidator validator ) {
        List<DataValidator> list = validators.get( entityName );

        if ( list == null )
        {
            list = new ArrayList<DataValidator>();
            validators.put( entityName, list );
        }

        list.add( validator );
    }

    public void addValidator( String entityName, DataValidator... dataValidators ) {
        for ( DataValidator t : dataValidators )
        {
            addValidator( entityName, t );
        }
    }

}
